package com.ibm.cps.messagequeue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Random;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Throwables;
import com.ibm.interfaces.IMessageClient;
import com.ibm.util.ErrorCode;
import com.ibm.util.SimpleQueue;
import com.ibm.util.exception.CPSException;

public class MQTTMessageQueueClient implements IMessageClient, MqttCallback {

	private MqttClient client;
	private String brokerUrl;
	private MqttConnectOptions conOpt;
	private String password;
	private String tenantName;
	private SimpleQueue queue;
	private static Logger logger;

	public MQTTMessageQueueClient(String brokerHost, int port,
			String tenantName, String password) throws CPSException {
		logger = LoggerFactory.getLogger(MQTTMessageQueueClient.class);
		this.brokerUrl = "tcp://" + brokerHost + ":" + port;
		this.password = password;
		this.tenantName = tenantName;
		queue = new SimpleQueue();
		connectToServer(tenantName, password);

	}

	private void connectToServer(String tenantName, String password)
			throws CPSException {
		String tmpDir = System.getProperty("java.io.tmpdir");
		Random random = new Random();
		long currentTime = System.currentTimeMillis() + random.nextInt(10000);
		MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(
				tmpDir + currentTime);

		try {
			conOpt = new MqttConnectOptions();
			conOpt.setCleanSession(true);
			if (password != null) {
				conOpt.setPassword(this.password.toCharArray());
			}
			if (tenantName != null) {
				conOpt.setUserName(this.tenantName);
			}
			logger.info("broker url is " + brokerUrl);
			// one connection for a specified client id
			client = new MqttClient(this.brokerUrl, "client_CRL" + currentTime,
					dataStore);
			client.setCallback(this);
			client.connect(conOpt);
			logger.info("Connected to " + brokerUrl + " with client ID "
					+ client.getClientId());
		} catch (MqttException e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					"Can't connect to server " + brokerUrl + " successfully.");
		}
	}

	@Override
	public void publishMessage(String topicName, String message)
			throws CPSException {
		publishMessage(topicName, message.getBytes());
	}

	@Override
	public void publishMessage(String topicName, byte[] message)
			throws CPSException {
		MqttMessage mqttMessage = new MqttMessage(message);
		try {
			client.publish(topicName, mqttMessage);
		} catch (MqttException e) {
			logger.error(Throwables.getStackTraceAsString(e));
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					e.getMessage());
		}
	}

	@Override
	public void subscribe(String topicName) throws CPSException {
		try {
			client.subscribe(topicName);
			logger.info("Subscribe topic " + topicName);
		} catch (MqttException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					e.getMessage());
		}
	}

	@Override
	public void unsubscribe(String topicName) throws CPSException {
		try {
			client.unsubscribe(topicName);
		} catch (MqttException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					e.getMessage());
		}
	}

	@Override
	public Collection<String> consumeMessage() throws CPSException {
		try {
			Collection<String> messages = new ArrayList<String>();
			messages.add((String) (queue.take()));
			return messages;
		} catch (InterruptedException e) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE,
					e.getMessage());
		}
	}

	@Override
	public void connectionLost(Throwable cause) {
		logger.warn(cause.getMessage());
	}

	@Override
	public void messageArrived(String topic, MqttMessage message)
			throws InterruptedException {
		logger.info("Add message " + topic + ": "
				+ new String(message.getPayload()));
		queue.put(new String(message.getPayload()));
	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {

	}

	@Override
	public void close() {
		try {
			client.close();
		} catch (MqttException e) {
			logger.info("Connection is closed.");
		}
	}
}
